package endpoint

import (
	"github.com/darwinOrg/go-common/constants"
	dgctx "github.com/darwinOrg/go-common/context"
	dglogger "github.com/darwinOrg/go-logger"
	dgmq "github.com/darwinOrg/go-mq"
	"github.com/go-mysql-org/go-mysql/canal"
	"github.com/go-mysql-org/go-mysql/mysql"
	"github.com/google/uuid"
	"github.com/juju/errors"
	"myc/global"
	"myc/metrics"
	"myc/model"
	"myc/util/alarm"
	"myc/util/logs"
	"myc/util/stringutil"
	"time"
)

type NatsEndpoint struct {
	host      string
	port      int
	username  string
	password  string
	poolSize  int
	mqAdapter dgmq.MqAdapter
}

func newNatsEndpoint() *NatsEndpoint {
	return &NatsEndpoint{
		host:     global.Cfg().NatsHost,
		port:     global.Cfg().NatsPort,
		username: global.Cfg().NatsUsername,
		password: global.Cfg().NatsPassword,
		poolSize: global.Cfg().NatsPoolSize,
	}
}

func (s *NatsEndpoint) Name() string {
	return "nats"
}

func (s *NatsEndpoint) Connect() error {
	adapter, err := dgmq.NewMqAdapter(&dgmq.MqAdapterConfig{
		Type:     dgmq.MqAdapterNats,
		Host:     s.host,
		Port:     s.port,
		Username: s.username,
		Password: s.password,
		PoolSize: s.poolSize,
		Timeout:  time.Second * 5,
		Group:    "myc",
	})
	if err != nil {
		logs.Errorf("dgmq.NewMqAdapter error: %v", err)
		return err
	}
	s.mqAdapter = adapter

	return nil
}

func (s *NatsEndpoint) Ping() error {
	return nil
}

func (s *NatsEndpoint) Close() {
	s.mqAdapter.Close()
}

func (s *NatsEndpoint) Consume(from mysql.Position, rows []*model.RowRequest) error {
	for _, row := range rows {
		rule, _ := global.RuleIns(row.RuleKey)
		if rule.TableColumnSize != len(row.Row) {
			logs.Warnf("%s schema mismatching", row.RuleKey)
			continue
		}

		metrics.UpdateActionNum(row.Action, row.RuleKey)

		err := s.doRuleConsume(row, rule)
		if err != nil {
			alarm.EmergencyAlarm(err)
			return err
		}
	}

	logs.Infof("%s 处理完成 %d 条数据", s.Name(), len(rows))
	return nil
}

func (s *NatsEndpoint) Stock(rows []*model.RowRequest) int64 {
	var sum int64
	for _, row := range rows {
		rule, _ := global.RuleIns(row.RuleKey)
		if rule.TableColumnSize != len(row.Row) {
			logs.Warnf("%s schema mismatching", row.RuleKey)
			continue
		}

		err := s.doRuleConsume(row, rule)
		if err != nil {
			logs.Errorf(errors.ErrorStack(err))
			alarm.EmergencyAlarm(err)
			break
		}
		sum++
	}

	return sum
}

func (s *NatsEndpoint) doRuleConsume(req *model.RowRequest, rule *global.Rule) error {
	kvm := rowMap(req, rule, false)

	resp := map[string]interface{}{}
	resp["action"] = req.Action
	resp["timestamp"] = req.Timestamp
	resp["data"] = encodeValue(rule, kvm)
	resp[constants.TraceId] = stringutil.UUID()

	if rule.ReserveRawData && canal.UpdateAction == req.Action {
		old := oldRowMap(req, rule, false)
		resp["raw"] = encodeValue(rule, old)
	} else {
		resp["raw"] = ""
	}

	ctx := &dgctx.DgContext{TraceId: uuid.NewString()}
	err := s.mqAdapter.Publish(ctx, rule.NatsTopic, resp)
	if err != nil {
		dglogger.Errorf(ctx, "Pub Message 错误 | topic: %s | error: %v", rule.NatsTopic, err)
		return err
	}

	dglogger.Infof(ctx, "topic: %s, message: %+v", rule.NatsTopic, resp)
	return nil
}
